package org.yumeko.kafka.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@RestController
public class KafkaVariousMethodsController {

    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;

    /**
     * 方法1: 指定主题和消息内容
     */
    @GetMapping("send/method1")
    public String method1(@RequestParam("message") String message) {
        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic-spring", "[方法1] " + message);

        // 添加回调处理
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("[方法1] 消息发送成功 - Topic: " + metadata.topic() +
                        ", Partition: " + metadata.partition() +
                        ", Offset: " + metadata.offset());
            } else {
                System.err.println("[方法1] 消息发送失败: " + ex.getMessage());
                ex.printStackTrace();
            }
        });

        return "Message sent using method 1";
    }

    /**
     * 方法2: 指定主题、分区和消息内容
     */
    @GetMapping("send/method2")
    public String method2(@RequestParam("message") String message) {
        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic-spring", 0, "[方法2] " + message);

        // 添加回调处理
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("[方法2] 消息发送成功 - Topic: " + metadata.topic() +
                        ", Partition: " + metadata.partition() +
                        ", Offset: " + metadata.offset());
            } else {
                System.err.println("[方法2] 消息发送失败: " + ex.getMessage());
                ex.printStackTrace();
            }
        });

        return "Message sent using method 2";
    }

    /**
     * 方法3: 指定主题、键和消息内容
     */
    @GetMapping("send/method3")
    public String method3(@RequestParam("message") String message) {
        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic-spring", 1, "[方法3] " + message);

        // 添加回调处理
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("[方法3] 消息发送成功 - Topic: " + metadata.topic() +
                        ", Partition: " + metadata.partition() +
                        ", Offset: " + metadata.offset());
            } else {
                System.err.println("[方法3] 消息发送失败: " + ex.getMessage());
                ex.printStackTrace();
            }
        });

        return "Message sent using method 3";
    }

    /**
     * 方法4: 使用ProducerRecord对象
     */
    @GetMapping("send/method4")
    public String method4(@RequestParam("message") String message) {
        ProducerRecord<Integer, String> record = new ProducerRecord<>(
                "topic-spring",
                2,
                999L, // timestamp
                2,    // key
                "[方法4] " + message
        );

        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(record);

        // 添加回调处理
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("[方法4] 消息发送成功 - Topic: " + metadata.topic() +
                        ", Partition: " + metadata.partition() +
                        ", Offset: " + metadata.offset() +
                        ", Timestamp: " + metadata.timestamp());
            } else {
                System.err.println("[方法4] 消息发送失败: " + ex.getMessage());
                ex.printStackTrace();
            }
        });

        return "Message sent using method 4";
    }

    /**
     * 方法5: 同步发送消息
     */
    @GetMapping("send/method5")
    public String method5(@RequestParam("message") String message) {
        try {
            SendResult<Integer, String> result = kafkaTemplate.send("topic-spring", "[方法5] " + message).get();

            RecordMetadata metadata = result.getRecordMetadata();
            System.out.println("[方法5] 消息发送成功 - Topic: " + metadata.topic() +
                    ", Partition: " + metadata.partition() +
                    ", Offset: " + metadata.offset());

            return "Message sent synchronously using method 5";
        } catch (InterruptedException | ExecutionException e) {
            System.err.println("[方法5] 消息发送失败: " + e.getMessage());
            e.printStackTrace();
            return "Failed to send message";
        }
    }

    /**
     * 方法6: 发送消息并flush
     */
    @GetMapping("send/method6")
    public String method6(@RequestParam("message") String message) {
        kafkaTemplate.send("topic-spring", "[方法6] " + message);
        kafkaTemplate.flush();

        System.out.println("[方法6] 消息已发送并刷新");
        return "Message sent and flushed using method 6";
    }

    /**
     * 方法7: 使用 Callback 回调处理发送结果
     */
    @GetMapping("send/method7")
    public String method7(@RequestParam("message") String message) {
        // 注意：KafkaTemplate 不直接支持 Callback 参数，需要使用原始的 KafkaProducer
        // 这里展示的是概念代码，实际使用时需要通过其他方式实现
        ProducerRecord<Integer, String> record = new ProducerRecord<>(
                "topic-spring", 2, 3, "[方法7] " + message);

        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(record);

        // 使用 whenComplete 实现类似 Callback 的功能
        future.whenComplete(new java.util.function.BiConsumer<SendResult<Integer, String>, Throwable>() {
            @Override
            public void accept(SendResult<Integer, String> result, Throwable ex) {
                if (ex == null) {
                    RecordMetadata metadata = result.getRecordMetadata();
                    System.out.println("[方法7] 消息发送成功 - Topic: " + metadata.topic() +
                            ", Partition: " + metadata.partition() +
                            ", Offset: " + metadata.offset());
                } else {
                    System.err.println("[方法7] 消息发送失败: " + ex.getMessage());
                    ex.printStackTrace();
                }
            }
        });

        return "Message sent using method 7";
    }

    /**
     * 方法8: 发送到默认主题
     */
    @GetMapping("send/method8")
    public String method8(@RequestParam("message") String message) {
        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.sendDefault("[方法8] " + message);

        future.whenComplete((result, ex) -> {
            if (ex == null) {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("[方法8] 消息发送成功 - Topic: " + metadata.topic() +
                        ", Partition: " + metadata.partition() +
                        ", Offset: " + metadata.offset());
            } else {
                System.err.println("[方法8] 消息发送失败: " + ex.getMessage());
                ex.printStackTrace();
            }
        });

        return "Message sent using method 8";
    }

    /**
     * 方法9: 批量发送消息
     */
    @GetMapping("send/method9")
    public String method9(@RequestParam("count") int count) {
        List<CompletableFuture<SendResult<Integer, String>>> futures = new ArrayList<>();

        // 创建多条消息
        for (int i = 0; i < count; i++) {
            String message = "[方法9] 批量消息 " + (i + 1);
            CompletableFuture<SendResult<Integer, String>> future =
                    kafkaTemplate.send("topic-spring", message);
            futures.add(future);
        }

        // 处理每条消息的发送结果
        for (int i = 0; i < futures.size(); i++) {
            final int index = i;
            futures.get(i).whenComplete((result, ex) -> {
                if (ex == null) {
                    RecordMetadata metadata = result.getRecordMetadata();
                    System.out.println("[方法9] 消息发送成功 - Topic: " + metadata.topic() +
                            ", Partition: " + metadata.partition() +
                            ", Offset: " + metadata.offset() +
                            ", Message: " + (index + 1));
                } else {
                    System.err.println("[方法9] 消息发送失败 (" + (index + 1) + "): " + ex.getMessage());
                    ex.printStackTrace();
                }
            });
        }

        return "Batch messages sent using method 9, count: " + count;
    }
}